package com.rocketmq.demo.base.producer;

import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.util.concurrent.TimeUnit;

/**
 * @ProjectName: rocketmq-demo
 * @Package: com.rocketmq.demo.base.producer
 * @ClassName: AsyncProducer
 * @Author: mac
 * @Description: 发送异步消息
 * @Date: 2021/4/1 10:48 上午
 * @Version: 1.0
 */
public class AsyncProducer {


    public static void main(String[] args) {
        //1.实例化消息生产者
        DefaultMQProducer producer = new DefaultMQProducer("group1");

        //设置 nameservice的地址
        producer.setNamesrvAddr("192.168.31.67:9876;192.168.31.126:9876;192.168.31.209:9876");

        try {
            //2.启动producer
            producer.start();

            //设置异步失败重试次数
            producer.setRetryTimesWhenSendAsyncFailed(0);

            //创建消息,并指定topic ,tag 和消息体
            for (int i = 0; i < 10; i++) {
                // 参数1:消息主题Topic 参数2:消息tag  参数3:消息类容
                Message msg = new Message("base", "Tag2", ("hello world" + 1).getBytes());
                //发送消息 sendCallback接收异步返回的结果的回调
                producer.send(msg, new SendCallback() {
                    /**
                     * 消息发送成功回调
                     * @param sendResult
                     */
                    @Override
                    public void onSuccess(SendResult sendResult) {
                        System.out.println("消息发送结果:" + sendResult.toString());
                    }

                    /**
                     * 消息发送失败回调
                     * @param throwable
                     */
                    @Override
                    public void onException(Throwable throwable) {
                        System.out.println("消息发送异常:" + throwable.toString());
                    }
                });

                TimeUnit.SECONDS.sleep(1);
            }
        } catch (MQClientException | RemotingException | InterruptedException e) {
            e.printStackTrace();
        } finally {
            //如果不发消息,关闭producer 实例
            producer.shutdown();
        }

    }

}
